package com.htdata.iiot;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.ResultSet;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.List;
import java.util.Random;

import org.apache.log4j.Logger;

import com.alibaba.fastjson.JSONObject;
import com.htdata.iiot.api.pojo.DeviceNumber;
import com.htdata.iiot.api.pojo.StreamingData;
import com.htdata.iiot.api.pojo.StreamingMetricData;
import com.htdata.iiot.db.DbConnectionPool;
import com.htdata.iiot.db.JDBCUtils;

/**
 * 格力实时数据
 * @author SunMy
 *
 */
public class Main8 {
	private static Logger logger = Logger.getLogger(Main8.class);
	private static DbConnectionPool pool;
	private static String topic;
	private static String tenantIds;
	private static int times=60;
	private static String zk;
	public static void main(String[] args) {

		for(String arg:args){
			String[] split = arg.split("=");
			switch (split[0]) {
			case "topic":
				topic = split[1];
				Configs.KafkaConfig.TOPIC=topic;
				break;
			case "zk":
				zk = split[1];
				System.setProperty("htiiot.zookeeper", zk);
				break;
			case "tids":
				tenantIds = split[1];
				break;
			case "times":
				times = Integer.parseInt(split[1]);
				break;

			default:
				break;
			}
		}
		pool  = DbConnectionPool.getPool();
		List<StreamingData> sList = new ArrayList<StreamingData>();
		SendDataToKafka sendDataToKafka = new SendDataToKafka();
		// 1.根据租户ID获取设备
		String sql1 = "SELECT tenant_id ,create_time as time, id,serial_number FROM `thing_device` "
				+ "WHERE deleted=0 AND type=0 ";
		
			sql1+=" and `tenant_id` in("+tenantIds+") ";
		
		System.out.println(sql1);
		Connection conn = pool.getConnection();
		PreparedStatement ps = null;
		ResultSet rs = null;
		ResultSet executeQuery = null;
		PreparedStatement prepareStatement = null;
		try {
			ps = conn.prepareStatement(sql1);
			
			rs = ps.executeQuery();
			while (rs.next()) {
				int deviceId = rs.getInt("id");
				int tenant_id = rs.getInt("tenant_id");
				StreamingData streamingData = new StreamingData();
				streamingData.setDid(deviceId);
				streamingData.setTid(tenant_id);
				// 2.根据设备获取测点
				String sql2 = "SELECT "
						+ "t1.id AS id,"
						+ "t2.`name` AS NAME "
						+ "FROM `thing_device_component_rel` t1,"
						+ " `thing_data_stream` t2 ,"
						+ "  thing_component t3 "
						+ "WHERE t1.`component_id` = t3.`id` "
						+ "  AND t3.deleted = 0 "
						+ "  AND t2.`component_id` = t3.`id` "
						+ "  AND t1.deleted = 0 "
						+ "  AND t1.device_id = ? "
						+ "  AND t2.deleted = 0 "
						+ "  AND t2.tenant_id = ?  ";
				prepareStatement = conn.prepareStatement(sql2);
				prepareStatement.setInt(1, deviceId);
				prepareStatement.setInt(2, tenant_id);
				executeQuery = prepareStatement.executeQuery();
				List<StreamingMetricData> mList = new ArrayList<StreamingMetricData>();
				while (executeQuery.next()) {
					String name = executeQuery.getString("name");
					long mId = executeQuery.getLong("id");
					StreamingMetricData metricData = new StreamingMetricData();
					metricData.setK(name);
					metricData.setDn(new DeviceNumber(tenant_id, deviceId, mId).toHexString());
					mList.add(metricData);
				}
				streamingData.setData(mList);
				if(mList.size()>0){
					sList.add(streamingData);
				}
			}

		} catch (SQLException e) {
			e.printStackTrace();
		} finally {
			JDBCUtils.close(ps, rs, conn);
			JDBCUtils.close(prepareStatement, executeQuery, null);
		}
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		logger.info(sList.size());
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		System.out.println("********");
		
		long i = 0;
		long mi = 0;
		long avg = sList.size()/(times*2);
		long startTime = System.currentTimeMillis()-1;
		long st = System.currentTimeMillis();
		while (true) {
			st = System.currentTimeMillis();
			// 3.生成数据
			for (StreamingData sData : sList) {
				long ts = System.currentTimeMillis();
				List<StreamingMetricData> data = sData.getData();
				for (StreamingMetricData mData : data) {
					switch (mData.getK()) {
					case "status":
						mData.setV(1000);
						break;
					case "AirVelocity":// 空速度
						mData.setV(50);
						break;
					case "backSpeed":// 回速度
						mData.setV(45);
						break;
					case "CuttingTimes":// 每分钟剪切次数
						mData.setV(3 + new Random().nextInt(4));
						break;
					case "flywheelDropRate":// 飞轮降速率
						mData.setV(0);// 计算
						break;
					case "focalLength":// 透镜焦距
						mData.setV(125);
						break;
					case "highSpeed":// 高速
						mData.setV(1000);
						break;
					case "InitialSpeed":// 飞轮初始转速
						mData.setV(300);
						break;
					case "MaleWeighStrength":// 公称力
						mData.setV(50);
						break;
					case "pressure":// 辅助气体气压//折弯机压力//冲床气源气压

						if (sData.getDid() == 86869) {// 激光切割机
							mData.setV(0.85 + new Random().nextInt(5) * 0.01);
						} else if (sData.getDid() == 87626) {// 折弯机
							mData.setV(10 + new Random().nextInt(5) * 0.1);
						} else if (sData.getDid() == 87632) {// 剪板机
							// mData.setV(1200);
						} else if (sData.getDid() == 87636) {// 冲床
							mData.setV(1 + new Random().nextInt(5) * 0.01);
						}

						break;
					case "RearSpeed":// 飞轮降速后转速
						mData.setV(300);
						break;
					case "speed":// 切速
						mData.setV(6);
						break;
					case "temperature":// 油温
						mData.setV(35 + new Random().nextInt(25));
						break;
					case "thickness":// 钢板厚度
						mData.setV(2);
						break;
					case "workSpeed":// 工作速度
						mData.setV(9);
						break;

					default:
						mData.setV(Math.ceil(new Random().nextDouble() * 10000) * 0.01);
						break;
					}
					mData.setTs(ts);
					mi++;
				}
				sData.setTs(ts);
				sendDataToKafka.send(sData.getDid(),buildResult(sData));
				i++;
				if(i%avg==0){
					//线程休息
					try {
						logger.info("中场休息");
						Thread.sleep(500);
					} catch (InterruptedException e) {
						e.printStackTrace();
					}
				}
			}
			long endTime = System.currentTimeMillis();
			long useTime = endTime - startTime;
			logger.info("写入 " + i + " 条数据,累计测点数："+mi+"，平均速率 " + (i * 1000 / useTime)+"，测点速率 " + (mi * 1000 / useTime));
			if(endTime<(st+times*1000)){
				try {
					Thread.sleep(st+times*1000-endTime);
				} catch (InterruptedException e) {
					e.printStackTrace();
				}
			}

		}

	}

	private static String buildResult(StreamingData sData) {
		return JSONObject.toJSONString(sData).toString();
	}

}
